#
# MediaDart, Copyright (C) 2006,2008 CRS4.
# All rights reserved.  Email: mediadart@crs4.it   Web: www.mediadart.org
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of:
#     The GNU Lesser General Public License as published by the Free
#     Software Foundation; either version 2.1 of the License, or (at
#     your option) any later version. The text of the GNU Lesser
#     General Public License is included with this library in the
#     file LICENSE.TXT.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the file
# LICENSE.TXT for more details.
#
# Derived from
#
import os
import sys
import types
import inspect
from types import TupleType, ListType, DictType, StringType, UnicodeType
from json import dumps, loads
from ConfigParser import NoSectionError, NoOptionError
from twisted.internet import defer, reactor
from twisted.python.failure import Failure
from twisted.internet.protocol import ClientCreator
from twisted.web.client import  HTTPClientFactory, HTTPPageGetter
import txamqp
from mediadart import log
from mediadart.mqueue import xmlspec
from mediadart.config import Configurator
from txamqp.protocol import AMQClient
from txamqp.client import TwistedDelegate
from txamqp.queue import Closed
from txamqp.content import Content

# set to log content of every message processed
VERBOSE = 0    

# used to identify a service when the service class does not have a _id attribute
# an id with the right syntax is obtained with the function "uuid.uuid1().get_urn()"
DUMMY_ID='urn:uuid:f021fe31-cbce-4039-b8a0-1d7e093c2d1b'

class RPCError(Exception):
    def __init__(self, arg):
        self.at = self.text = None
        if type(arg) in [StringType, UnicodeType]:
            Exception.__init__(self, arg)
        elif type(arg) == DictType:
            Exception.__init__(self, arg['message'], arg['at'], arg['text'])
        else:
            Exception.__init__(self, 'Received unrecognized error type')

    def __str__(self):
        ret = '%s: %s' % (self.__class__.__name__, str(self.args[0]))
        if len(self.args) > 1:
            ret += ' (at %s)' % ':'.join(map(str, self.args[1:]))
        return ret

# An AMQClient singleton for all server classes
_client_singleton = None
_aLock = defer.DeferredLock()

def startMQ(mq_class, args, broker_ip, broker_port, username, password, vhost):
    """ 
        Creates a connection to the broker, attaches the resulting AMQClient
        object to mq_class,  calls setup on this class.
        An existing connection is reused, so that all servers in an instance
        of twisted share the same connection.
        If a connection attempt is pending, it waits to use the connection
        after it is established.
    """
    def start():
        mq_class(*args)._setup()

    def cb0(lock):
        global _client_singleton
        if not _client_singleton: 
            log.debug('%s: making new connection' % (mq_class.__name__,))
            spec_path = os.path.dirname(os.path.abspath(inspect.getfile(MQServer)))
            specfile = os.path.join(spec_path, xmlspec)
            spec = txamqp.spec.load(specfile)
            delegate = TwistedDelegate()
            #log.debug('connectTCP %s:%s' % (broker_ip, broker_port))
            d = ClientCreator(reactor, AMQClient, TwistedDelegate(), vhost, spec).connectTCP(broker_ip, int(broker_port))
            d.addCallbacks(cb1, cb_err, callbackArgs=[lock], errbackArgs=[lock])
        else:
            lock.release()
            d = None
            start()
        return d


    def cb1(client, lock):
        #log.debug('startMQ(%s): got the client: %d' % (mq_class.__name__, id(client)))
        d = client.authenticate(username, password)
        d.addCallbacks(cb2, cb_err, callbackArgs=[client, lock], errbackArgs=[lock])
        return d

    def cb2(result, client, lock):
        global _client_singleton
        #log.debug('startMQ(%s) authentication ok' % (mq_class.__name__,))
        _client_singleton = client
        lock.release()
        start()

    def cb_err(failure, lock):
        global _client_singleton
        _client_singleton = None
        lock.release()
        raise RPCError( 'error: %s' % str(failure))

    global _aLock
    #log.info('connecting to broker %s:%s, user=%s, vhost=%s' % (broker_ip, broker_port, username, vhost))
    _get_config_options(mq_class.__name__.upper(), writelog=False)  # to check that the default is a positive number
    d = _aLock.acquire()
    d.addCallback(cb0)

class MQServer(object):
    def __init__(self):
        if not _client_singleton:
            raise  RPCError('connection to broker missing: use startMQ to create servers')
        self.__client = _client_singleton
        self.__qname =  self.__class__.__name__
        self.__tag = 'mq%d' % id(self)
        self.__chid = None
        self.__outstanding = 0   # number of requests in execution
        self.__concurrency, self.__dispatch_delay, self.__exclusive = _get_config_options(self.__qname.upper())
        self._description = None
        #log.debug('opening queue=%s, tag=%s' % (self.__qname, self.__tag))

    # see older versions for _setup with inlineCallbacks. A little shorter, but
    # exceptions are harder to read and control. No sugar, thanks!
    def _setup(self):
        def cb1(channel_id):
            self._channel = channel_id
            self._channel.channel_open().addCallbacks(cb2, self._setup_error)

        def cb2( result ):
            self._log_msg(result)
            self._channel.exchange_declare(exchange="amq.direct", type="direct").addCallbacks(cb3, self._setup_error)

        def cb3( result ):
            self._log_msg(result)
            self._channel.queue_declare(queue=self.__qname, durable=True, auto_delete=True, exclusive=self.__exclusive).addCallbacks(cb4, self._setup_error)

        def cb4( result ):
            self._log_msg(result)
            self._channel.queue_bind(queue=self.__qname, exchange="amq.direct", 
                  routing_key=self.__qname).addCallbacks(cb5, self._setup_error)

        def cb5( result ):
            self._log_msg(result)
            self._channel.basic_consume(queue=self.__qname, no_ack=True, 
                 consumer_tag=self.__tag).addCallbacks(cb6, self._setup_error)

        def cb6( result ):
            "Get the queue corresponding to tag and attach _next_msg to it"
            self._log_msg(result)
            self.__client.queue(self.__tag).addCallbacks(lambda queue: self._next_msg(queue), self._setup_error)

        self.__chid = len(self.__client.channels) + 100   # unique-leave low values for txamqp
        self.__client.channel(self.__chid).addCallbacks(cb1, self._setup_error)

    def _setup_error(self, failure):
        log.error('ERROR: failed to create proxy: %s' % failure)

    def _next_msg(self, queue, only_when_full=0):
        #if outstanding is too small, the next message is already being read
        if only_when_full and self.__outstanding < (self.__concurrency - 1):
            return
        if self.__outstanding < self.__concurrency:
            self.__outstanding += 1
            queue.get().addCallbacks(self._handle_msg, self._handle_internal_err, 
                            callbackArgs=[queue], errbackArgs=[queue])
        
    def _handle_internal_err(self, failure, queue):
        self.__outstanding -= 1
        if failure.check(Closed):
            log.info('channel closed')
        else:
            log.error('error: %s' % failure)
            self._next_msg(queue)

    def _handle_msg(self, msg, queue):
        self._log_msg(msg)
        replyq = msg.content.properties.get('reply to',None) 
        #log.debug('LLLLLLLL: %s' % replyq)
        if not msg.content or not replyq:
            self.__outstanding -= 1
            log.error('not an mq message')
        else:
            reactor.callLater(self.__dispatch_delay, self._dispatch, msg, queue, replyq)
        self._next_msg(queue)

    def _process_args(self, params):
        positional_params = []
        kw_params = {}
        if type(params) in [ListType, TupleType]:
            positional_params = params
        elif type(params) == DictType:
            positional_keys = [x for x in params.keys() if x.isdigit()]
            positional_keys.sort(key=int)
            positional_params = [params[x] for x in positional_keys]
            non_positional = [x for x in params.keys() if x not in positional_keys]
            kw_params.update([(x, params[x]) for x in non_positional])
        return positional_params, kw_params

    def _dispatch(self, msg, queue, replyq):
        try:
            req = loads(msg.content.body)
        except Exception, e:
            log.debug('MQServer.dispatch: json decoding error: %s' % str(e))
        if req['version'] != '1.1':
            log.debug('WARNING: version=%s (version supported is 1.1)' % req['version'])
        methodname = req['method'].replace('.', '_')
        args, kwargs = self._process_args(req['params'])
        log.info('executing %s' % methodname)
        msg_id = msg.content.properties['correlation id']
        handler = getattr(self, 'mq_%s' % methodname, None)
        if not handler:
            self._handle_resp(Failure(Exception('MQServer dispatch: %s: unrecognized method' % methodname)), msg_id, queue, replyq, 'error')
        else:
            d = defer.Deferred()
            kwargs_str = dict([(str(x[0]), x[1]) for x in kwargs.items()])
            log.debug('calling %s with args=%s, kwargs=%s' % (handler.__name__, args, kwargs_str))
            d.addCallback(lambda _ : handler(*args, **kwargs_str))
            d.addCallbacks(callback=self._handle_resp, 
                           callbackArgs=(msg_id, queue, replyq,'result'),
                           errback=self._handle_resp, 
                           errbackArgs=(msg_id, queue, replyq, 'error'),)
            d.callback(None)

    def _format_response(self, resp_type, msg_id, result):
        """Return the a json-encoded dictionary

        result is coded as is, otherwise it is first rendered as string
        """
        ret = None
        def get_failure_repr(f):
            if type(f.value) in [types.DictType, types.ListType, types.TupleType]:
                return f.value
            else:
                return f.getErrorMessage()

        encoders = {
         'error': [get_failure_repr, lambda x: "result non json encodable"],
         'result':[lambda x: x, str, lambda x: "result non json encodable"],
        }
        for encoder in encoders[resp_type]:
            try:
                ret = dumps({'version':'1.1', resp_type:encoder(result), 'id':msg_id})
                break
            except:
                log.debug('_format_respons failed: %s' % encoder(result))
                pass
        return ret

    def _handle_resp(self, result, msg_id, queue, replyq, resp_type):
        self.__outstanding -= 1
        self._next_msg(queue, 1)
        ret = self._format_response(resp_type, msg_id, result)
        log.debug('request completed: returning(%s): %s\n### over #####' % (resp_type, ret[:128].rstrip()))
        if resp_type == 'error':
            log.debug(result.getTraceback())
        self._send_reply(ret, msg_id, replyq)

    def _send_reply(self, value, msg_id, queueid):
        content = Content(value)
        content.properties['correlation id'] = msg_id
        self._log_content(content)
        if queueid.startswith('http://'):
            def log_ok(page):
                log.debug('Response set successfully to %s' % queueid)

            def log_error(reason):
                log.error('error sending response to %s: %s' % (queueid, reason))

            log.debug('Replying directly to %s' % queueid)
            factory = doHTTP(queueid, value)
            factory.deferred.addCallbacks(log_ok, log_error)
        else:
            self._channel.basic_publish(exchange="amq.direct", content=content, routing_key=queueid)


    def mq_system_describe(self):
        "return the service description"
        if not self._description:
            #log.debug('describing')
            self._description = {'name': self.__class__.__name__,
                   'sdversion': '1.0',
                   'id': getattr(self, '_id', DUMMY_ID),
                   }
            docstring = inspect.getdoc(self)
            if docstring:
                self._description['summary'] = docstring
            #log.debug('describing docstring')
            procs = []
            self._description['procs'] = []
            for name, func in inspect.getmembers(self, inspect.ismethod):
                if name[:3] != 'mq_':
                    continue
                d = {'name': name[3:] }
                docstring = inspect.getdoc(func)
                if docstring:
                    d['summary'] = docstring
                #params = []
                #arg_names = inspect.getargspec(func)
                #for arg in arg_names.args[1:]:
                #    params.append({'name':arg, 'type':'any'})
                #d['params'] = params
                #d['return'] = 'any'
                self._description['procs'].append(d)
        return self._description

    def mq_ping(self):
        "returns basic stats on the server"
        return getattr(self, '_stats', 'ok')

    if VERBOSE:
        def _log_msg(self, msg):
            log.debug('Message %s' % msg.method)
            log.debug(' content: %s' % msg.content)
            log.debug(' args: %s' % str(msg.fields))
            if msg.content:
                log.debug(' prop.: %s' % msg.content.properties)
            log.debug('\n' + '<'*80 + '\n')

        def _log_content(self, content):
            log.debug('content: %s' % content)
            log.debug(' properties: %s' % content.properties)
            log.debug('\n%s\n' % ('<'*80))
    else:
        def _log_msg(self, msg): pass
        def _log_content(self, content): pass

#utility function
def _get_config_options(section='', writelog=True):
    c = Configurator()
    concurrency = -1
    dispatch_delay = -1
    try:
        exclusive = c.getboolean(section, 'only_one_server')
    except (NoOptionError, NoSectionError), e:
        exclusive = False
    try:
        concurrency = c.getint(section, 'concurrency_level')
    except (NoOptionError, NoSectionError), e:
        pass
    if concurrency < 0:
        concurrency = c.getint('MQUEUE', 'default_concurrency_level')
    if concurrency <= 0:
        raise RPCError('Invalid concurrency level in configuration')
    try:
        dispatch_delay = c.getfloat(section, 'dispatch_delay')
    except (NoOptionError, NoSectionError), e:
        pass
    if dispatch_delay < 0:
        dispatch_delay = c.getfloat('MQUEUE', 'default_dispatch_delay')
    if dispatch_delay < 0:
        raise RPCError('Invalid dispatch_delay in configuration')
    if writelog:
        log.info('%16s: concurrency=%2d, dispatch_delay=%5.2f exclusive=%s' % (section, concurrency, dispatch_delay, exclusive))
    return concurrency, dispatch_delay, exclusive


def doHTTP(url, data='', method='POST', headers = {}, followRedirect=False):
    factory = HTTPClientFactory(url=str(url), method=method, postdata=data,
               headers=headers, followRedirect=followRedirect)
    reactor.connectTCP(factory.host, factory.port, factory)
    return factory

